﻿using MassTransit;
using MassTransit.ActiveMqTransport;
using MassTransit.Configuration;
using MessageBus.Const;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;


namespace MessageBus
{
    public class MessageBusClient
    {
        public BrokerInfo BrokerInfo { get; protected set; }
        public BrokerType BrokerType { get; protected set; }

        /// <summary>
        /// <监听队列,对应处理方法>
        /// </summary>
        private Dictionary<string, Action<IReceiveEndpointConfigurator>> _receiveDic;
        private Action<IReceiveEndpointConfigurator> _configureEndpoint = null;
        private IBusControl _busControl;
        private BusHandle _busHandler;

        private ConcurrentDictionary<string, ISendEndpoint> _sendEndpoints;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="busType">消息总线的实现方式</param>
        /// <param name="connectionInfo"></param>
        /// <param name="receiveDic"></param>
        public MessageBusClient(
            BrokerType busType,
            BrokerInfo brokerInfo,
            Dictionary<string, Action<IReceiveEndpointConfigurator>> receiveDic = null)
        {
            BrokerType = busType;
            BrokerInfo = brokerInfo;
            _receiveDic = receiveDic ?? new Dictionary<string, Action<IReceiveEndpointConfigurator>>(0);
            _sendEndpoints = new ConcurrentDictionary<string, ISendEndpoint>();
        }

        public async ValueTask StartAsync()
        {
            _busControl ??= CreateBusControl();
            _busHandler ??= await _busControl.StartAsync();
        }

        public async ValueTask StopAsync()
        {
            await _busHandler.StopAsync();
        }

        public IBusControl CreateBusControl()
        {
            return this.BrokerType switch
            {
                BrokerType.Memory => Bus.Factory.CreateUsingInMemory(cfg =>
                    {
                        foreach (KeyValuePair<string, Action<IReceiveEndpointConfigurator>> kv in _receiveDic)
                        {
                            cfg.ReceiveEndpoint(kv.Key, kv.Value);
                        }
                    }),
                BrokerType.RabbitMQ => Bus.Factory.CreateUsingRabbitMq(cfg =>
                    {
                        cfg.Host(BrokerInfo.IP, BrokerInfo.Port, BrokerInfo.VirtualHost, cnf =>
                        {
                            cnf.Username(BrokerInfo.User);
                            cnf.Password(BrokerInfo.Pwd);
                        });
                        foreach (KeyValuePair<string, Action<IReceiveEndpointConfigurator>> kv in _receiveDic)
                        {
                            cfg.ReceiveEndpoint(kv.Key, kv.Value);
                        }
                    }),
                BrokerType.ActiveMQ => Bus.Factory.CreateUsingActiveMq(cfg =>
                    {
                        cfg.Host(BrokerInfo.IP, BrokerInfo.Port, cnf =>
                        {
                            cnf.Username(BrokerInfo.User);
                            cnf.Password(BrokerInfo.Pwd);
                        });
                        foreach (KeyValuePair<string, Action<IReceiveEndpointConfigurator>> kv in _receiveDic)
                        {
                            cfg.ReceiveEndpoint(kv.Key, kv.Value);
                        }
                    }),
                _ => null,
            };
        }

        public async ValueTask<bool> SendAsync(string sencQueue, object msg)
        {
            bool exist = _sendEndpoints.TryGetValue(sencQueue, out ISendEndpoint sendEndpoint);
            if (!exist)
            {
                string typeStr = null;
                typeStr = GetUrlHeader(BrokerType);

                sendEndpoint = await _busControl.GetSendEndpoint(new Uri($"{typeStr}://{BrokerInfo.IP}:{BrokerInfo.Port}/{sencQueue}"));
                exist = _sendEndpoints.TryAdd(sencQueue, sendEndpoint);
                if (!exist) { return false; }
            }
            await sendEndpoint.Send(msg);
            return true;
        }

        private string GetUrlHeader(BrokerType type)
        {
            return type switch
            {
                BrokerType.ActiveMQ => "activemq",
                BrokerType.RabbitMQ => "amqps",
                _ => throw new NotImplementedException(),
            };
        }
    }

    public class Msg { public string Text { get; set; } }
    public class MsgConsumer : IConsumer<Msg>
    {
        public async Task Consume(ConsumeContext<Msg> context)
        {
            Console.WriteLine(context.Message.Text);
        }
    }
}